package com.shujia.streaming

import org.apache.spark.sql.streaming.OutputMode
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object Demo05StructuredStream {
  def main(args: Array[String]): Unit = {

    // 创建SparkSession
    val spark: SparkSession = SparkSession
      .builder()
      .appName("Demo05StructuredStream")
      .master("local[2]")
      .config("spark.sql.shuffle.partitions", "2")
      .getOrCreate()

    // nc -lk 8888
    val linesDF: DataFrame = spark
      .readStream
      .format("socket") // 以socket作为source
      .option("host", "master")
      .option("port", 8888)
      .load()
    import spark.implicits._
    import org.apache.spark.sql.functions._

    // wordCount
    // DSL
    // 默认会记录状态 因为延时比较低（大约在100ms左右） 所以看起来像来一条数据处理一条
    //    linesDF
    //      .select(explode(split($"value", ",")) as "words")
    //      .groupBy($"words")
    //      .agg(count("*") as "cnt")
    //      .writeStream
    //
    //      /**
    //       * outputMode 有三种模式
    //       * Append 不能用于带聚合的操作 只能适用于简单的查询
    //       * Complete 会将所有的数据都输出 只能用于带聚合的操作
    //       * Update 会将新增或修改的数据进行输出 如果没有聚合函数 则相当于Append
    //       */
    //      .outputMode(OutputMode.Complete())
    //      .format("console") // sink到控制台
    //      .start()
    //      .awaitTermination()
    // SQL
    linesDF.createOrReplaceTempView("word_count")

    // 流批合一
    spark.sql(
      """
        |SELECT  t1.word
        |        ,count(*) as cnt
        |from (
        |    select  explode(split(value,",")) as word
        |    from word_count
        |) t1 group by t1.word
        |""".stripMargin)
      .writeStream
      .outputMode(OutputMode.Complete()) // 可以使用DF的API进行处理
      .foreachBatch((df, l) => {
        df.write.format("jdbc")
          .option("url", "jdbc:mysql://master:3306/bukong?useSSL=false")
          .option("dbtable", "word_cnt")
          .option("user", "root")
          .option("password", "123456")
          .mode(SaveMode.Overwrite) // Spark SQL在以overwrite模式写MySQL时会影响MySQL中的表结构
          .save()
      })
      .start()
      .awaitTermination()


  }

}
